package com.qy566.gps.pulsar.demo;


import org.apache.pulsar.client.api.*;

import java.util.Date;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * @author wangnaiqi
 * @description TODO
 * @date 2023/11/8
 */
public class ProducerDemo {
//    public static void main(String[] args) throws PulsarClientException, InterruptedException {
//        // 获取pulsar的客户端对象
//        ClientBuilder clientBuilder = PulsarClient.builder();
//        clientBuilder.serviceUrl("pulsar://10.0.0.27:32524");
//        PulsarClient client = clientBuilder.build();
//        // 通过客户端创建生产者的对象
//        Producer<User> producer = client.newProducer(JSONSchema.of(User.class)).topic("persistent://my-tenant/my-namespace/my-topic").create();
//        // 发送消息
//        int i = 0;
//        while (i < 200) {
//            User user = new User();
//            user.setAge(11);
//            user.setName("张三");
//            producer.send(user);
//            Thread.sleep(10);
//            i++;
//        }
//        // 释放资源
//        producer.close();
//        client.close();
//    }

    public static void main(String[] args) throws PulsarClientException {
        //1.创建Pulsar的客户端对象
        PulsarClient pulsarClient = PulsarClient.builder()
                .serviceUrl("pulsar://10.0.0.27:30980")
                .authentication(AuthenticationFactory.token("eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.UUlFnlvhcPHMWPNAebzWwt6R71MxKgUIKZEZA3WeJ7d74AY2O5vFit6CcF4QyMyJri8EAzLDidkrkxmRwJ0egzqmEZLYDtKtAwBSKBrucoOzsZ-gTu_escnt_Qtzpa7QGcE2KX7wpW3-2BuolBrnmaipQqoZrq54Zlse2qA9bJgObJmXAMvIAO6tmZEHqtsWCzZV1jfXjiHE8SlESeVQ8TpvoyUzD6R2Bx36LEGRACNYxs_k0E2FO0cSflWB3sCS7mOE8ksqstNLdiN54ECfIk3AlQ0ahkvRk8b4Gweq0Vv2E87NBIFVYpYKyDFiUTlVcRaCr0RQk96J6T2qVsoTfg"))
                .build();

        //2.通过客户端构建生产者的对象
        //可以通过命令创建：./pulsar-admin topics create persistent://toto_pulsar_t/toto_pulsar_n/t_topic1
        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
                .topic("persistent://gpsTenant/deviceNamespace/deviceCarBindingTopic")
                .enableBatching(false)
                .create();

        //3.使用生产者发送数据
        producer.newMessage().value("{\n" +
                "        \"deviceCode\": \"000000012345\",\n" +
                "        \"carNo\": \"鲁A12345\",\n" +
                "        \"companyBindOperateEnum\":\"ADD\"\n" +
                "    }").sendAsync();
        System.out.println("数据生产完成...");

        //4.释放资源
        producer.close();
        pulsarClient.close();
    }
}
